package com.home.fansea.listener;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @Description: TODO
 **/
@Component
public class RedisMQListener implements StreamListener<String, MapRecord<String, String, Object>> {
 
    private static final Logger log = LoggerFactory.getLogger(RedisMQListener.class);
 
    // 创建一个线程池
    /*private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            10, 20, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());*/
    private static final ExecutorService threadPoolExecutor = Executors.newSingleThreadExecutor();

    @Override
    public void onMessage(MapRecord message) {
        // 异步处理消息
        /*threadPoolExecutor.execute(() -> {
            System.out.println(Thread.currentThread().getName() + ":接收到的消息：" + message.getId() + ";" + JSON.toJSONString(message.getValue()));
        });*/
        threadPoolExecutor.execute(new MqttMSgHandle(message));
 
    }

    @AllArgsConstructor
    public class MqttMSgHandle implements Runnable{

        //这边实现读取消息队列创建订单的操作
        private MapRecord message;
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + ":接收到的消息：" + message.getId() + ";" + JSON.toJSONString(message.getValue()));
            try {
                //处理MQTT的数据

                //1、解析MQTT数据判断是否达到报警阈值

                //

            }catch (Exception e){
                log.error("出现订单异常",e);
                handlePendingList();
            }
        }

        private void handlePendingList() {
            while (true){
                try {
                    //处理MQTT的数据

                }catch (Exception e){
                    log.error("出现pending-list订单异常",e);
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                    }
                }
            }
        }
    }
}